package com.tiantang.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @description:功能描述 HBase Java API 测试
 * @author: liujinkun
 * @date 2018年5月17日
 * @version:1.0
 */
public class APITest {

	// 日志工具
	private static Logger logger = LoggerFactory.getLogger(APITest.class);

	// HBase的配置对象，该对象包含了HBase和Hadoop的配置信息
	private Configuration conf = null;

	// 连接HBase的对象
	private Connection connection = null;

	/**
	 * 初始化配置对象
	 * 
	 * @return
	 */
	public Configuration initConfiguration() {
		if (conf == null) {
			conf = HBaseConfiguration.create();
		}
		return conf;
	}

	/**
	 * 在每一个单元测试方法之前，初始化连接
	 */
	@Before
	public void before() {
		initConfiguration();
		if (connection == null) {
			try {
				connection = ConnectionFactory.createConnection(conf);
			} catch (IOException e) {
				logger.info("创建hbase连接出现异常，e={}", e);
				e.printStackTrace();
			}
		}
	}

	/**
	 * 获取HBase中标的管理对象
	 * 
	 * @return
	 * @throws IOException
	 */
	public Admin getAdmin() throws IOException {
		return connection.getAdmin();
	}

	/**
	 * 
	 * @Title: tableExists
	 * @description:功能描述 判断HBase中表是否存在
	 * @param tableName 表名
	 * @return true：存在，false：不存在
	 * @author:liujinkun
	 * @date 2018年5月17日
	 */
	public boolean tableExists(String tableName) {
		try {
			Admin admin = getAdmin();
			return admin.tableExists(TableName.valueOf(tableName));
		} catch (IOException e) {
			logger.info("获取HBase的表管理对象出现异常，e={}", e);
			e.printStackTrace();
		}
		return false;
	}
	
	/**
	 * 
	 * @Title: createTable  
	 * @description:功能描述 创建表
	 * @param tableName 表名
	 * @param columnFamily 列族
	 * @return   true：创建成功，false：创建失败
	 * @author:liujinkun
	 * @date 2018年5月19日
	 */
	public boolean createTable(String tableName,String... columnFamilys){
		if(tableExists(tableName)){
			logger.info("表已经存在");
			return false;
		}
		try {
			Admin admin=getAdmin();
			//创建表需要通过表描述器进行创建
			HTableDescriptor hTableDescriptor=new HTableDescriptor(TableName.valueOf(tableName));
			
			//列族描述器
			for (String columnFamily : columnFamilys) {
				HColumnDescriptor hcolumnDescriptor=new HColumnDescriptor(Bytes.toBytes(columnFamily));
				hTableDescriptor.addFamily(hcolumnDescriptor);
			}
			
			admin.createTable(hTableDescriptor);
			return true;
		} catch (IOException e) {
			logger.info("创建表出现异常，e={}",e);
			e.printStackTrace();
			return false;
		}
	}
	
	/**
	 * 
	 * @Title: deleteTable  
	 * @description:功能描述 删除表
	 * @param tableName 表名
	 * @return   true:删除成功,false:删除失败
	 * @author:liujinkun
	 * @date 2018年5月20日
	 */
	public boolean deleteTable(String tableName){
		if(!tableExists(tableName)){
			logger.info("表不存在");
			return false;
		}
		try {
			Admin admin=getAdmin();
			//删除表之前需要先卸载表（disable）,否则会出现异常：org.apache.hadoop.hbase.TableNotDisabledException
			admin.disableTable(TableName.valueOf(tableName));
			admin.deleteTable(TableName.valueOf(tableName));
			return true;
		} catch (IOException e) {
			logger.info("删除表出现异常,e={}",e);
			e.printStackTrace();
			return false;
		}
	}
	
	/**
	 * 
	 * @Title: addRowData  
	 * @description:功能描述 添加数据
	 * @param tableName 表名
	 * @param rowKey 行键
	 * @param columnFamily 列族
	 * @param column 列名
	 * @param value 值
	 * @return   true：添加成功，false：添加失败
	 * @author:liujinkun
	 * @date 2018年5月20日
	 */
	public boolean addRowData(String tableName,String rowKey,String columnFamily,String column,String value){
		try {
			//获取表对象
			Table table=connection.getTable(TableName.valueOf(tableName));
			//将要添加的数据封装成一个PUT对象
			Put put=new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
			//将数据添加到表中
			table.put(put);
			return true;
		} catch (IOException e) {
			logger.info("添加数据出现异常，e={}",e);
			e.printStackTrace();
			return false;
		}
	}
	
	/**
	 * 
	 * @Title: batchDeleteRows  
	 * @description:功能描述 批量删除数据
	 * @param tableName 表名
	 * @param rowKeys 行键
	 * @return   true：删除成功，false：删除失败
	 * @author:liujinkun
	 * @date 2018年5月20日
	 */
	public boolean batchDeleteRows(String tableName,String...rowKeys){
		if (rowKeys.length <= 0) {
			logger.info("请传入rowKey");
			return false;
		}
		try {
			//获取表对象
			Table table=connection.getTable(TableName.valueOf(tableName));
			//删除表数据需要将删除的数据封装成一个Delete对象
			List<Delete> deletes=new ArrayList<Delete>();
			for (String rowKey : rowKeys) {
				Delete delete=new Delete(Bytes.toBytes(rowKey));
				deletes.add(delete);
			}
			//删除数据
			table.delete(deletes);
			return true;
		} catch (IOException e) {
			logger.info("批量删除数据出现异常，e={}",e);
			e.printStackTrace();
			return false;
		}
	}
	
	/**
	 * 
	 * @Title: getAllData  
	 * @description:功能描述 获取表的所有的数据
	 * @param tableName   表名
	 * @author:liujinkun
	 * @date 2018年5月20日
	 */
	public void getAllData(String tableName){
		try {
			Table  table=connection.getTable(TableName.valueOf(tableName));
			Scan scan=new Scan();
			//ResultScanner对象表示的是多个行键对应数据的集合
			ResultScanner resultScanner = table.getScanner(scan);
			//Result对象表示一个行键对应的数据，一个行键可以包含多个单元格
			for (Result result : resultScanner) {
				//Cell对象表示一个单元格中对应的数据
				List<Cell> cells = result.listCells();
				logger.info("====从result中获取的行键为：row={}",Bytes.toString(result.getRow()));
				for (Cell cell : cells) {
					logger.info("单元格的行键为：rowKey={}",Bytes.toString(CellUtil.cloneRow(cell)));
					logger.info("单元格的列族为：columnFamily={}",Bytes.toString(CellUtil.cloneFamily(cell)));
					logger.info("单元格的列名为：column={}",Bytes.toString(CellUtil.cloneQualifier(cell)));
					logger.info("单元格的值为：column={}",Bytes.toString(CellUtil.cloneValue(cell)));
				}
			}
		} catch (IOException e) {
			logger.info("获取所有的数据出现异常,e={}",e);
			e.printStackTrace();
		}
	}
	
	/**
	 * 
	 * @Title: getRowData  
	 * @description:功能描述 根据行键获取一行数据
	 * @param tableName 表名
	 * @param rowKey 行键
	 * @return   Result 封装了rowKey所对应的数据
	 * @author:liujinkun
	 * @date 2018年5月20日
	 */
	public Result getRowData(String tableName,String rowKey){
		try {
			Table table=connection.getTable(TableName.valueOf(tableName));
			//Get 对象封装查询条件
			Get get=new Get(Bytes.toBytes(rowKey));
			// 设置最大版本号
			get.setMaxVersions();
			//当行键在数据库中不存在时，查询到的Result对象并不为null，但result.listCells()的结果为null
			return table.get(get);
		} catch (IOException e) {
			logger.info("查询某一行的数据出现异常，e={}",e);
			e.printStackTrace();
			return null;
		}
	}
	
	/**
	 * 
	 * @Title: getRowQualifiler  
	 * @description:功能描述  查询指定行指定列的数据
	 * @param tableName 表名
	 * @param rowKey 行键
	 * @param columnFamily 列族
	 * @param column 列
	 * @return   Result
	 * @author:liujinkun
	 * @date 2018年5月20日
	 */
	public Result getRowQualifiler(String tableName,String rowKey,String columnFamily,String column){
		try {
			Table table=connection.getTable(TableName.valueOf(tableName));
			Get get=new Get(Bytes.toBytes(rowKey));
			get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
			return table.get(get);
		} catch (IOException e) {
			logger.info("查询指定行指定列的数据出现异常,e={}",e);
			e.printStackTrace();
			return null;
		}
	}
	
	@Test
	public void testGetRowQualifier(){
		Result result=this.getRowQualifiler("student", "1001", "info", "age");
		if(result == null){
			logger.info("没有获取到数据");
			return;
		}
		Cell[] cells = result.rawCells();
		for (Cell cell : cells) {
			logger.info("行键为：rowKey={}",Bytes.toString(CellUtil.cloneRow(cell)));
			logger.info("列族为：columnFamily={}",Bytes.toString(CellUtil.cloneFamily(cell)));
			logger.info("列名为：column={}",Bytes.toString(CellUtil.cloneQualifier(cell)));
			logger.info("列值为：value={}",Bytes.toString(CellUtil.cloneValue(cell)));
			logger.info("时间戳：timestamp={}",cell.getTimestamp());
		}
	}
	
	@Test
	public void testGetRowData(){
		//当行键在数据库中不存在时，查询到的Result对象并不为null，但result.listCells()的结果为null
		Result result=this.getRowData("student", "1001");
		if(result == null){
			logger.info("没有获取到数据");
			return;
		}
		List<Cell> cells = result.listCells();
		for (Cell cell : cells) {
			logger.info("行键为：rowKey={}",Bytes.toString(CellUtil.cloneRow(cell)));
			logger.info("列族为：columnFamily={}",Bytes.toString(CellUtil.cloneFamily(cell)));
			logger.info("列名为：column={}",Bytes.toString(CellUtil.cloneQualifier(cell)));
			logger.info("列值为：value={}",Bytes.toString(CellUtil.cloneValue(cell)));
			logger.info("时间戳：timestamp={}",cell.getTimestamp());
		}
	}
	
	@Test
	public void testGetAllData(){
		this.getAllData("student");
	}
	
	@Test
	public void testBatchDeleteRows(){
		boolean result=this.batchDeleteRows("student", "1001");
		logger.info("批量删除数据的结果为：result={}",result);
	}
	@Test
	public void testAddRowData(){
//		boolean result=this.addRowData("student", "1001", "info", "name", "Tom");
		boolean result=this.addRowData("student", "1001", "info", "age", "29");
//		boolean result=this.addRowData("student", "1002", "info", "name", "TianTang");
		logger.info("添加数据的结果为：result={}",result);
	}
	
	@Test
	public void testDeleteTable(){
		boolean result=this.deleteTable("ns_weibo:user_relation");
		logger.info("删除表的结果为：result={}",result);
	}
	
	@Test
	public void testCreateTable(){
		boolean result=this.createTable("test","family1","family2");
		logger.info("创建表的结果为：result={}",result);
	}

	@Test
	public void testTableExists() {
		boolean result = this.tableExists("student");
		logger.info("表是否存在：reuslt={}", result);
	}

	@Test
	public void run() throws IOException {
		Configuration conf = HBaseConfiguration.create();
		Connection connection = ConnectionFactory.createConnection(conf);
		Admin admin = connection.getAdmin();
		boolean result = admin.tableExists(TableName.valueOf("student"));
		logger.info("启动成功,查询表是否存在的结果为：result={}", result);
	}

}
